package com.tbit.uqbike.service.redis.impl;

import com.tbit.uqbike.service.redis.RedisService;
import com.tbit.uqbike.util.Pair;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.dao.DataAccessException;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.core.*;
import org.springframework.data.redis.serializer.RedisSerializer;
import redis.clients.util.JedisByteHashMap;

import java.util.*;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;


public class RedisServiceImpl implements RedisService {
    @Autowired
    private RedisTemplate<String, String> redisTemplate;

    private ConcurrentHashMap<String, Integer> runCnt = new ConcurrentHashMap<>();

    public RedisServiceImpl() {
    }

    private void IncRun(String name) {
        if (!runCnt.containsKey(name)) {
            runCnt.put(name, 1);
        } else {
            runCnt.put(name, runCnt.get(name) + 1);
        }
    }

    @Override
    public String getRunCnt() {
        StringBuilder sb = new StringBuilder();
        Set<Map.Entry<String, Integer>> set = runCnt.entrySet();
        for (Map.Entry<String, Integer> kv : set) {
            sb.append(String.format("%s:%d,", kv.getKey(), kv.getValue()));
            kv.setValue(0);
        }
        return sb.toString();
    }

    @Override
    public void set(String key, String value) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();
        valueOperation.set(key, value);
        IncRun("set");
    }

    @Override
    public void set(String hashId, String key, String value) {
        HashOperations<String, String, String> hashOperations = redisTemplate.opsForHash();
        hashOperations.put(hashId, key, value);
//        redisTemplate.expire(hashId, 34560000, TimeUnit.SECONDS);
        IncRun("hset");
    }

    @Override
    public void persist(String key) {
        redisTemplate.persist(key);
    }

    @Override
    public void Key_Del(String key) {
        redisTemplate.delete(key);
    }

    @Override
    public String get(String key) {
        ValueOperations<String, String> valueOperation = redisTemplate.opsForValue();
        String ret = valueOperation.get(key);
        IncRun("get");
        return ret;
    }

    @Override
    public String get(String hashId, String key) {
        HashOperations<String, String, String> hashOperations = redisTemplate.opsForHash();
        String ret = hashOperations.get(hashId, key);
        IncRun("hget");
        return ret;
    }

    @Override
    public Map<String, String> getAll(String hashId) {
        HashOperations<String, String, String> hashOperations = redisTemplate.opsForHash();
        Map<String, String> ret = hashOperations.entries(hashId);
        IncRun("hgetall");
        return ret;
    }

    @Override
    public List<Object> Hash_BatGetAll(final List<String> keys) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer proSer = redisTemplate.getHashKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                for (String keyStr : keys) {
                    byte[] key = hashIdSer.serialize(keyStr);
                    connection.hGetAll(key);
                }
                return connection.closePipeline();
            }
        });
        IncRun("Hash_BatGetAll");
        return result;
    }

    @Override
    public String getFirstItem(String key) {
        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        String str = listOperations.index(key, 0);
        IncRun("getFirstItem");
        return str;
    }

    @Override
    public String remoteFirstItem(String key) {
        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        String str = listOperations.leftPop(key);
        IncRun("remoteFirstItem");
        return str;
    }

    @Override
    public List<String> LRange(String key, int start, int end) {
        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        List<String> list = listOperations.range(key, start, end);
        IncRun("LRange");
        return list;
    }

    @Override
    public Long LDelValue(String key, long count, Object obj) {
        ListOperations<String, String> listOperations = redisTemplate.opsForList();
        Long l = listOperations.remove(key, 0, obj);
        IncRun("LDelValue");
        return l;
    }

    @Override
    public List<Object> BatSetHashValue(final HashMap<String, String> dic, final String proName) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer proSer = redisTemplate.getHashKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                Set<Map.Entry<String, String>> set = dic.entrySet();
                for (Map.Entry<String, String> entry : set) {
                    byte[] key = hashIdSer.serialize(entry.getKey());
                    byte[] pro = proSer.serialize(proName);
                    byte[] value = valueSer.serialize(entry.getValue());
                    connection.hSet(key, pro, value);
                }
                return connection.closePipeline();
            }
        });
        IncRun("BatSetHashValue");
        return result;
    }

    @Override
    public List<String> BatGetHashValue(final List<String> keys, final String proName) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer proSer = redisTemplate.getHashKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                for (String keyStr : keys) {
                    byte[] key = hashIdSer.serialize(keyStr);
                    byte[] pro = proSer.serialize(proName);
                    connection.hGet(key, pro);
                }
                return connection.closePipeline();
            }
        });
        IncRun("BatGetHashValue");
        List<String> strRet = new LinkedList<>();
        for (Object obj : result) {
            if (null != obj && obj instanceof byte[]) {
                strRet.add(valueSer.deserialize((byte[]) obj).toString());
            } else {
                strRet.add(null);
            }
        }
        return strRet;
    }

    @Override
    public List<String> BatGetValue(final List<String> keys) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                for (String keyStr : keys) {
                    byte[] key = hashIdSer.serialize(keyStr);
                    connection.get(key);
                }
                return connection.closePipeline();
            }
        });
        IncRun("BatGetValue");
        List<String> strRet = new LinkedList<>();
        for (Object obj : result) {
            if (null != obj && obj instanceof byte[]) {
                strRet.add(valueSer.deserialize((byte[]) obj).toString());
            } else {
                strRet.add(null);
            }
        }
        return strRet;
    }

    @Override
    public List<Long> BatGetListLen(final List<String> keys) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                for (String keyStr : keys) {
                    byte[] key = hashIdSer.serialize(keyStr);
                    connection.lLen(key);
                }
                return connection.closePipeline();
            }
        });
        IncRun("BatGetListLen");
        List<Long> strRet = new LinkedList<>();
        for (Object obj : result) {
            if (null != obj && obj instanceof Long) {
                strRet.add((Long) obj);
            } else {
                strRet.add(0L);
            }
        }
        return strRet;
    }

    @Override
    public List<Object> BatSetHashValue(final HashMap<String, List<Map.Entry<String, String>>> dic) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer proSer = redisTemplate.getHashKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                Set<Map.Entry<String, List<Map.Entry<String, String>>>> set = dic.entrySet();
                for (Map.Entry<String, List<Map.Entry<String, String>>> entry : set) {
                    byte[] key = hashIdSer.serialize(entry.getKey());
                    List<Map.Entry<String, String>> kvList = entry.getValue();
                    for (Map.Entry<String, String> kv : kvList) {
                        byte[] pro = proSer.serialize(kv.getKey());
                        byte[] value = valueSer.serialize(kv.getValue());
                        connection.hSet(key, pro, value);
                    }
                }
                return connection.closePipeline();
            }
        });
        IncRun("BatSetHashValueV2");
        return result;
    }

    @Override
    public List<Object> Hash_BatSetValue(final List<Pair<String, List<Pair<String, String>>>> data) {
        final RedisSerializer hashIdSer = redisTemplate.getKeySerializer();
        final RedisSerializer proSer = redisTemplate.getHashKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                for (Pair<String, List<Pair<String, String>>> item : data) {
                    byte[] hashId = hashIdSer.serialize(item.getKey());
                    List<Pair<String, String>> kvList = item.getValue();
                    for (Pair<String, String> kv : kvList) {
                        byte[] pro = proSer.serialize(kv.getKey());
                        byte[] value = valueSer.serialize(kv.getValue());
                        connection.hSet(hashId, pro, value);
                    }
                }
                return connection.closePipeline();
            }
        });
        IncRun("Hash_BatSetValue");
        return result;
    }

    @Override
    public List<Object> Set_BatSetKey(final String setKey, final List<String> items) {
        final RedisSerializer keySer = redisTemplate.getKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getValueSerializer();
        final List<Object> result = redisTemplate.execute(new RedisCallback<List<Object>>() {
            @Override
            public List<Object> doInRedis(RedisConnection connection) throws DataAccessException {
                connection.openPipeline();
                byte[] key = keySer.serialize(setKey);
                for (String item : items) {
                    byte[] value = valueSer.serialize(item);
                    connection.sAdd(key, value);
                }
                return connection.closePipeline();
            }
        });
        IncRun("Set_BatSetKey");
        return result;
    }

    @Override
    public HashMap<String, String> Hash_FormatResult(JedisByteHashMap byteHashMap) {
        final RedisSerializer proSer = redisTemplate.getHashKeySerializer();
        final RedisSerializer valueSer = redisTemplate.getHashValueSerializer();
        Set<Map.Entry<byte[], byte[]>> set = byteHashMap.entrySet();
        HashMap<String, String> resulHash = new HashMap<>();
        for (Map.Entry<byte[], byte[]> kv : set) {
            resulHash.put(proSer.deserialize(kv.getKey()).toString(), valueSer.deserialize(kv.getValue()).toString());
        }
        return resulHash;
    }

    @Override
    public Set<String> Set_GetAllItem(String setKey) {
        SetOperations<String, String> setOperations = redisTemplate.opsForSet();
        Set<String> set = setOperations.members(setKey);
        IncRun("Set_GetAllItem");
        return set;
    }
}  